package com.lx.cotroller;

import cn.hutool.core.util.RandomUtil;
import com.aliyuncs.CommonRequest;
import com.aliyuncs.CommonResponse;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.http.MethodType;
import com.aliyuncs.profile.DefaultProfile;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lx.bean.User;
import com.lx.service.UserService;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.lang.model.SourceVersion;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

@RestController
@RequestMapping("/user")
public class UserController {
    @Autowired
    private UserService userService;

    @PostMapping("register")
    public Map register(@RequestBody User user) throws Exception{
        System.out.println("user = " + user);
        User user1 = userService.selectUserByPhone(user.getPhone());
        System.out.println("user1 = " + user1);
        Map map = new HashMap();
        if(user1 != null && user.getValidate().equals(user1.getValidate())){
            userService.updateUserByPhone(user);
            map.put("status","success");
        }else {
            map.put("status","failure");
        }
        return map;
    }
    @PostMapping("send")
    public Map send (@RequestBody User user) throws Exception{
        user.setValidate(RandomUtil.randomString (4));
        producer(user);//调用生产者方法
        userService.addUser(user);
        return new HashMap(){
            {put("status","success");}
        };
    }
    protected void producer(User user) throws Exception{
        Map<String,Object> conf = new HashMap<>();
        conf.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop10:9092");
        conf.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        conf.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);

        KafkaProducer<String, String> producer = new KafkaProducer<>(conf);

        ObjectMapper mapper = new ObjectMapper();
        String s = mapper.writeValueAsString(user);//{"phone":"p123","password":"p123456","validate":"vabc"}
        producer.send(new ProducerRecord<>("topica",s));

        producer.close();
    }

}
